import tkinter
import tkinter.messagebox as messagebox
import confluent_kafka
from kafka import KafkaAdminClient, KafkaConsumer, TopicPartition
from kafka.admin import NewTopic
from confluent_kafka.admin import AdminClient, ConfigResource
from collections import defaultdict
import time

main_windows = tkinter.Tk()
main_windows.title("天璇之kafka")
width = 750
height = 600
main_windows.geometry(f'{width}x{height}')


# print(main_windows.keys())

# 获取文本内容

def acked(err, msg):
    if err is not None:
        print(f"Failed to deliver message: {err}")
    else:
        print(f'End of partition reached {msg.topic()} [{msg.partition()}] at offset {msg.offset()}')
        conn_label.config(text="连接成功")


def conn_kafka():
    kafka_ip = kafkaEntry1.get()
    kafka_port = kafkaEntry2.get()
    kafka_topic = topicEntry1.get()
    producer = confluent_kafka.Producer({'bootstrap.servers': f'{kafka_ip}:{kafka_port}'})
    producer.produce(topic=kafka_topic, key='key', value='value', callback=acked)
    producer.flush()  # 等待消息发送完成


def topic_total():
    kafka_ip = kafkaEntry1.get()
    kafka_port = kafkaEntry2.get()
    consumer = confluent_kafka.Consumer({
        'bootstrap.servers': f'{kafka_ip}:{kafka_port}',
        'group.id': "my_python_group_id",
        'auto.offset.reset': 'earliest'
    })
    topics = consumer.list_topics().topics
    print(topics)
    second_label2.config(text=f'{kafka_ip}:{kafka_port}共计{len(topics)}个topic')


def topic_show():
    kafka_ip = kafkaEntry1.get()
    kafka_port = kafkaEntry2.get()
    consumer = confluent_kafka.Consumer({
        'bootstrap.servers': f'{kafka_ip}:{kafka_port}',
        'group.id': "my_python_group_id",  # 消费者组ID
        'auto.offset.reset': 'earliest'  # 从最早的消息开始读取
    })
    topics = consumer.list_topics().topics
    for topic in sorted(topics):
        mu_text.insert(tkinter.END, topic + '\n')


def topic_info():
    kafka_ip = kafkaEntry1.get()
    kafka_port = kafkaEntry2.get()
    topic_name = second_entry.get()

    # configure的用不了
    admin_client = KafkaAdminClient(bootstrap_servers=f'{kafka_ip}:{kafka_port}')
    try:
        topic_description_list = admin_client.describe_topics([topic_name])
        # for循环获取输入的topic信息
        for topic_dict in topic_description_list:
            for key, value in topic_dict.items():
                if key == "topic":
                    mu_text.insert(tkinter.END, f"topic名称: {value}" + '\n')
                elif key == "is_internal":
                    mu_text.insert(tkinter.END, f"topic是否内置: {value}" + '\n')
                elif key == "partitions":
                    for topic_partitions_dict in topic_dict[key]:
                        mu_text.insert(tkinter.END,
                                       f"Partition ID: {topic_partitions_dict['partition']}, "
                                       f"Leader: {topic_partitions_dict['leader']}, "
                                       f"Replicas: {topic_partitions_dict['replicas']}, "
                                       f"ISR: {topic_partitions_dict['isr']}, "
                                       f"offline_replicas: {topic_partitions_dict['offline_replicas']}" +
                                       '\n')
    except Exception as e:
        mu_text.insert(tkinter.END, f"连接{kafka_ip}:{kafka_port}异常: {e}" + '\n')

    admin_config_client = AdminClient({'bootstrap.servers': f'{kafka_ip}:{kafka_port}'})
    topic_resource = ConfigResource(ConfigResource.Type.TOPIC, topic_name, )
    try:
        configs = admin_config_client.describe_configs([topic_resource])
        for _, config in configs.items():
            mu_text.insert(tkinter.END, "打印配置信息：" + '\n')
            for key, value in config.result().items():
                mu_text.insert(tkinter.END, f"{key}: {value}" + '\n')

    except Exception as e:
        mu_text.insert(tkinter.END, f"连接{kafka_ip}:{kafka_port}异常: {e}" + '\n')


def topic_create():
    kafka_ip = kafkaEntry1.get()
    kafka_port = kafkaEntry2.get()
    topic_name = second_entry.get()
    topic_fenqu = int(third_entry1.get())
    topic_fuben = int(third_entry2.get())
    # 创建 AdminClient 实例
    admin_client = KafkaAdminClient(bootstrap_servers=f'{kafka_ip}:{kafka_port}')
    new_topic = NewTopic(topic_name, num_partitions=topic_fenqu, replication_factor=topic_fuben)
    # 创建 topic
    try:
        admin_client.create_topics([new_topic])
        topics_set = admin_client.list_topics()
        if topic_name in topics_set:
            mu_text.insert(tkinter.END, f"创建topic: {topic_name} 成功" + '\n')
            second_label2.config(text=f"创建topic: {topic_name} 成功")
        else:
            mu_text.insert(tkinter.END, f"创建topic: {topic_name} 失败" + '\n')
            second_label2.config(text=f"创建topic: {topic_name} 失败")
    except Exception as e:
        mu_text.insert(tkinter.END, f"创建topic异常: {e}" + '\n')
        second_label2.config(text=f"创建topic: {topic_name} 异常")

    # 关闭 AdminClient
    admin_client.close()


def topic_delete():
    kafka_ip = kafkaEntry1.get()
    kafka_port = kafkaEntry2.get()
    topic_name = second_entry.get()
    if messagebox.askokcancel("确认", "是否确定要继续删除？"):
        # 创建 AdminClient 实例
        admin_client = KafkaAdminClient(bootstrap_servers=f'{kafka_ip}:{kafka_port}')
        admin_client.delete_topics(topics=[topic_name])
        # 检查是否成功删除主题
        topics_set = admin_client.list_topics()
        if topic_name not in topics_set:
            mu_text.insert(tkinter.END, f"删除topic: {topic_name} 成功" + '\n')
            second_label2.config(text=f"删除topic: {topic_name} 成功")
        else:
            mu_text.insert(tkinter.END, f"删除topic: {topic_name} 失败" + '\n')
            second_label2.config(text=f"删除topic: {topic_name} 失败")
    else:
        second_label2.config(text=f"取消删除{topic_name}")


def topic_read():
    kafka_ip = kafkaEntry1.get()
    kafka_port = kafkaEntry2.get()
    topic_name = second_entry.get()
    max_consume_num = int(second_entry3.get())
    consumer = KafkaConsumer(
        topic_name,
        bootstrap_servers=f'{kafka_ip}:{kafka_port}',
        auto_offset_reset='earliest',
        enable_auto_commit=True
    )
    consumer.poll(0)  # 非阻塞调用，立即返回
    mu_text.insert(tkinter.END, f"开始读取消息: {topic_name}" + '\n')
    for message in consumer:
        mu_text.insert(tkinter.END,
                       f"[{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))}] key={message.key} value={message.value.decode('utf-8')}")

    # # 订阅主题
    # consumer.subscribe([topic_name])
    # # 获取分区和 offset
    # partitions = [TopicPartition(topic_name, partition) for partition in consumer.partitions_for_topic(topic_name)]
    # end_offsets = consumer.end_offsets(partitions)
    # # 获取每个分区最大的消费数量
    # partition_offsets = {}
    # # for循环分区列表，如何offsets字典里没有这个分区，置为0
    # for partition in partitions:
    #     if not end_offsets[partition]:
    #         partition_offsets[partition.partition] = 0
    #         continue
    #     # 否则用offset减要看的条数
    #     offset = end_offsets[partition] - max_consume_num
    #     if offset < 0:
    #         # 分区的数据量小于最大的消费量
    #         # 因此从 0 开始消费即可
    #         offset = 0
    #         partition_offsets[partition.partition] = end_offsets[partition]
    #     else:
    #         partition_offsets[partition.partition] = max_consume_num
    #     consumer.seek(partition, offset)
    # # 所有分区加起来要消费的数据量
    # actual_max_consume_num = sum(partition_offsets.values())
    # if not actual_max_consume_num:
    #     # topic 数据为空
    #     mu_text.insert(tkinter.END, f"{topic_name} Empty" + '\n')
    # else:
    #     consumed = defaultdict(int)
    #     for m in consumer:
    #         if m.partition not in partition_offsets:
    #             continue
    #         consumed[m.partition] += 1
    #         if consumed[m.partition] > partition_offsets[m.partition]:
    #             if all(v >= partition_offsets[k] for k, v in consumed.items()):
    #                 # 如果全部分区都已经消费到了最大的数据量
    #                 # 则终止消费
    #                 break
    #             continue
    #         mu_text.insert(tkinter.END, f'{m}' + '\n')
    consumer.close()


def clear_text():
    mu_text.delete('1.0', tkinter.END)


# 控制最右侧位置
i = 7
bt1 = tkinter.Button(main_windows, text="topic总数", command=topic_total)
bt2 = tkinter.Button(main_windows, text="topic展示", command=topic_show)
bt3 = tkinter.Button(main_windows, text="topic信息", command=topic_info)
bt4 = tkinter.Button(main_windows, text="topic创建", command=topic_create)
bt5 = tkinter.Button(main_windows, text="topic删除", command=topic_delete)
bt6 = tkinter.Button(main_windows, text="topic读取", command=topic_read)
bt1.grid(row=0, column=1, padx=10)
bt2.grid(row=0, column=2, padx=10)
bt3.grid(row=0, column=3, padx=10)
bt4.grid(row=0, column=4, padx=10)
bt5.grid(row=0, column=5, padx=10)
bt6.grid(row=0, column=6, padx=10)

# 二行标签和输入框
second_label2 = tkinter.Label(main_windows, text="查询结果", anchor='w')
second_label2.grid(row=2, column=1, columnspan=3)
second_label1 = tkinter.Label(main_windows, text="输入topic:")
second_entry = tkinter.Entry(main_windows)
second_entry.insert('0', 'congyang_test')
second_label3 = tkinter.Label(main_windows, text="读取行数:", anchor='w')
second_entry3 = tkinter.Entry(main_windows, width=5)
second_entry3.insert(1, '5')
second_label1.grid(row=2, column=4, sticky=tkinter.W)
second_entry.grid(row=3, column=4, columnspan=2, sticky=tkinter.W)
second_label3.grid(row=2, column=6, sticky=tkinter.W)
second_entry3.grid(row=3, column=6, sticky=tkinter.W)

# 三行
third_label1 = tkinter.Label(main_windows, text="分区数:")
third_entry1 = tkinter.Entry(main_windows, width=5)
third_entry1.insert(0, "3")
third_label2 = tkinter.Label(main_windows, text="副本数:")
third_entry2 = tkinter.Entry(main_windows, width=5)
third_entry2.insert(0, "1")
third_label1.grid(row=4, column=4, sticky=tkinter.W)
third_entry1.grid(row=4, column=5, sticky=tkinter.W)
third_label2.grid(row=5, column=4, sticky=tkinter.W)
third_entry2.grid(row=5, column=5, sticky=tkinter.W)

# 右侧展示连接信息
addr_var = "172.16.100.72"
entry_var = tkinter.StringVar(value=addr_var)
kafkaLabel1 = tkinter.Label(main_windows, text="kafka地址:")
kafkaEntry1 = tkinter.Entry(main_windows, width=15, textvariable=entry_var)
kafkaLabel2 = tkinter.Label(main_windows, text="kafka端口:")
kafkaEntry2 = tkinter.Entry(main_windows, width=5)
kafkaEntry2.insert(0, "9092")
topicLabel1 = tkinter.Label(main_windows, text="topic名称:")
topicEntry1 = tkinter.Entry(main_windows, width=15)
topicEntry1.insert(0, 'congyang_test')
zkLabel1 = tkinter.Label(main_windows, text="zookeeper地址:")
zkEntry1 = tkinter.Entry(main_windows, width=15, textvariable=entry_var)
zkLabel2 = tkinter.Label(main_windows, text="zookeeper端口:")
zkEntry2 = tkinter.Entry(main_windows, width=5)
zkEntry2.insert(0, "2181")
conn_button = tkinter.Button(main_windows, text='测试连接', command=conn_kafka)
conn_label = tkinter.Label(main_windows, text="未连接", anchor='w')
kafkaLabel1.grid(row=0, column=i, sticky=tkinter.W)
kafkaEntry1.grid(row=0, column=i + 1, sticky=tkinter.W)
kafkaLabel2.grid(row=1, column=i, sticky=tkinter.W)
kafkaEntry2.grid(row=1, column=i + 1, sticky=tkinter.W)
topicLabel1.grid(row=2, column=i, sticky=tkinter.W)
topicEntry1.grid(row=2, column=i + 1, sticky=tkinter.W)
zkLabel1.grid(row=3, column=i, sticky=tkinter.W)
zkEntry1.grid(row=3, column=i + 1, sticky=tkinter.W)
zkLabel2.grid(row=4, column=i, sticky=tkinter.W)
zkEntry2.grid(row=4, column=i + 1, sticky=tkinter.W)
conn_button.grid(row=5, column=i, sticky=tkinter.W)
conn_label.grid(row=5, column=i + 1)

mu_button = tkinter.Button(main_windows, text="清屏", command=clear_text)
mu_text = tkinter.Text(main_windows)
scrollbar = tkinter.Scrollbar(main_windows)
scrollbar.config(command=mu_text.yview)
mu_text.config(yscrollcommand=scrollbar.set)
mu_button.grid(row=8, column=1)
mu_text.grid(row=9, column=1, columnspan=8, sticky='nsew')
scrollbar.grid(row=9, column=9, sticky='ns')

# print(bt.grid_info())

main_windows.mainloop()
